package top.zpengblog.kafkaprogram.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import top.zpengblog.kafkaprogram.consumer.KafKaConsumer;
import top.zpengblog.kafkaprogram.producer.KafKaProducer;

/**
 * @author linzp
 * @version 1.0.0
 * CreateDate 2020/11/19 19:54
 */
@RestController
public class KafkaController {

    @Autowired
    private KafKaProducer kafKaProducer;

    @Autowired
    private KafKaConsumer kafKaConsumer;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     *
     * @return
     */
    @PostMapping("/sendKafkaMsg")
    public String putMsg(@RequestParam String msg){

        kafKaProducer.produce(msg);
        return "success";
    }
    /**
     *
     * @return
     */
    @PostMapping("/sendKafkaMsg2")
    public String putMsg2(@RequestParam String msg){

        kafkaTemplate.send("kafkaMsg", msg);
        return "success"+System.currentTimeMillis();
    }

    /**
     *
     * @return
     */
    @PostMapping("/getKafkaMsg")
    public String getMsg(){
        kafKaConsumer.consume();
        return "success";
    }
}
